package lion.framework.core.task.cluster;

import java.util.List;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

import lion.dev.lang.MapJ;
import lion.framework.core.task.AutoTaskManager;
import lion.framework.core.task.ISchedule;
import lion.framework.core.task.builder.IScheduleBuilder;
import lion.framework.core.task.builder.ScheduleBuilderManager;

/**
 * 孤儿任务守护者<br/>
 * 孤儿任务：指那些执行任务的实例宕机后，没有任何实例来运行它的任务。<br/>
 * 该守护线程会3秒钟检查一次，如果发现有孤儿任务,则将一部分孤儿任务接替到本实例下。
 *
 * <pre>
 * time   2013年7月25日
 * mail   hl_0074@sina.com
 * </pre>
 *
 * @author lion
 */
public class OrphanScheduleRetriever extends TimerTask {

	private Map<String, ClusterInstance> clusterInstances;
	private AtomicBoolean start = new AtomicBoolean(false);
	private static Log log = LogFactory.getLog(OrphanScheduleRetriever.class);
	private static final String LOCK_NAME = "lock_find_orphan";

	public OrphanScheduleRetriever(Map<String, ClusterInstance> clusterInstances) {

		this.clusterInstances = clusterInstances;
	}

	public void start() {

		if (start.get()) { return; }
		start.getAndSet(true);

		Timer timer = new Timer("orphan-schedule-checker", true);
		timer.scheduleAtFixedRate(this, 0, 3000);
	}

	@Override
	public void run() {

		if (clusterInstances.isEmpty()) { return; }

		for (ClusterInstance instance : clusterInstances.values()) {

			if (!instance.getLock(LOCK_NAME)) {
				continue;
			}

			try {
				List<MapJ> targetids = instance.findOrphanSchedules();
				for (MapJ sc : targetids) {

					String id = sc.getString("id");
					String type = sc.getString("type");

					IScheduleBuilder builder = ScheduleBuilderManager.get(type);
					if (builder == null) {
						continue;
					}
					MapJ conf = builder.getScheduleConfig(id);
					ISchedule schedule = builder.build(conf);
					if (schedule == null) {
						continue;
					}
					if (instance.addSchedule(schedule)) {
						AutoTaskManager.addSchedule(schedule);
					}
					if (log.isDebugEnabled()) {
						log.debug("schedule " + schedule.getId() + " is retrived to " + instance.getId());
					}
				}
			} catch (Throwable e) {
				log.error(e);
			} finally {
				instance.releaseLock(LOCK_NAME);
			}
		}
	}

}
